package com.sojson.util.lock.impl;

import static com.sojson.util.lock.LockUtil.TIMEOUT_LONG;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sojson.config.exception.TransErrorCode;
import com.sojson.config.exception.TransException;
import com.sojson.util.StringUtil;
import com.sojson.util.lock.CustomLock;
import com.sojson.util.lock.DistributedLockUtilInterface;
import com.sojson.util.lock.LockUtil;
import com.sojson.util.zookeeper.ZookeeperUtil;
import com.sojson.util.zookeeper.ZookeeperUtilInterface;
import com.sojson.util.zookeeper.impl.ACustomWatcher;

import lombok.extern.slf4j.Slf4j;

/**
 * Zookeeper分布式锁工具类实现类
 * 
 * @author liu
 * @date 2020-11-15
 */
@Slf4j
public class DistributedLockZookeeperUtilImpl implements DistributedLockUtilInterface {

    private static DistributedLockZookeeperUtilImpl dis;
    /** Zookeeper工具类 */
    private ZookeeperUtilInterface zk = createZk();
    /** 锁在Zookeeper上的根目录 */
    final String root = "/" + LockUtil.CACHE_NAME_PREFIX_LOCK;
    /** 锁在Zookeeper上的根目录(为了减少拼接次数) */
    final String root_ = root + "/";
    /** root路径的长度+1 */
    private final int rootLength = root.length() + 1;
    /** 锁名称和锁索引的连接符(因为锁名称后续会由Zookeeper自动在后面拼上索引,要从所有名称的锁中准确取出当前名称的锁,所以加上这个用来取出锁名) */
    private final String SPLITSTR = "_lock_";

    private DistributedLockZookeeperUtilImpl() {
        // 创建一个与服务器的连接
        Stat stat;
        try {
            // 如果基础节点不存在就创建
            stat = zk.exists(root);
            if (stat == null) {
                // 创建根节点
                zk.create(root);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    /**
     * Zookeeper分布式锁监视类
     * 
     * @author liu
     * @date 2020-11-15
     */
    class CustomLockWatcher extends ACustomWatcher implements CustomLock {

        private final Logger log = LoggerFactory.getLogger(CustomLockWatcher.class);
        /** 当前锁(简称) */
        private String lockName;
        /** 当前锁(全称,格式为: lockName+SPLITSTR+由Zookeeper自动生成的索引) */
        private String myNode;
        /** 要等待的锁(全称,格式为: lockName+SPLITSTR+由Zookeeper自动生成的索引)(这个锁的内容执行完成后会执行当前锁的内容) */
        private String waitNode;
        /** 当这个数值为0的时候执行当前锁的内容(上一个锁执行完成后会删除节点,节点删除后会让这个对象的数值减一) */
        private CountDownLatch latch;

        public CustomLockWatcher(String lockName) {
            this.lockName = lockName;
        }

        /**
         * 监视器(回调方法)
         */
        @Override
        public void process(WatchedEvent event) {
            // 其他线程放弃锁的标志
            if (StringUtil.isNotBlankObject(latch)) {
                this.latch.countDown();
            }
        }

        /**
         * 获取默认锁(有获取锁的超时时间,有使用锁的超时时间)
         * 
         * @return
         */
        @Override
        public boolean defaultLock() {
            return defaultLock(LockUtil.TIMEOUT_USE, LockUtil.TIME_UNIT);
        }

        /**
         * 获取默认锁
         * 
         * @param timeOutUse    使用锁的超时时间(单位: 秒,不设置为30秒)
         * @return
         */
        @Override
        public boolean defaultLock(long timeOutUse) {
            return defaultLock(timeOutUse, LockUtil.TIME_UNIT);
        }

        /**
         * 获取默认锁
         * 
         * @param timeOutUse    使用锁的超时时间(单位: 秒,不设置为30秒)
         * @param unit          使用锁的超时时间单位
         * @return
         */
        @Override
        public boolean defaultLock(long timeOutUse, TimeUnit unit) {
            return lock(LockUtil.TIMEOUT_GET, timeOutUse, unit);
        }

        /**
         * 获取锁,如果获取到就结束,没有获取到就一直暂停直到抛异常或者程序结束
         */
        @Override
        public boolean lock() {
            return lock(TIMEOUT_LONG, TIMEOUT_LONG, null);
        }

        /**
         * 获取锁,如果获取到返回true,没有获取到返回false
         * 
         * @return
         */
        @Override
        public boolean tryLock() {
            return tryLock(LockUtil.TIMEOUT_GET, LockUtil.TIME_UNIT);
        }

        /**
         * 在给定的时间内获取锁,如果获取到返回true,超时或者没有获取到返回false
         * 
         * @param timeOutGet    获取锁的超时时间(单位: 秒,不设置为10秒)
         * @return
         */
        @Override
        public boolean tryLock(long timeOutGet) {
            return tryLock(timeOutGet, LockUtil.TIME_UNIT);
        }

        /**
         * 在给定的时间内获取锁,如果获取到返回true,超时或者没有获取到返回false
         * 
         * @param timeOutGet    获取锁的超时时间(单位: 秒,不设置为10秒)
         * @param unit          获取锁的超时时间单位
         * @return
         */
        @Override
        public boolean tryLock(long timeOutGet, TimeUnit unit) {
            return lock(timeOutGet, LockUtil.TIMEOUT_USE, LockUtil.TIME_UNIT);
        }

        /**
         * 在给定的时间内获取锁,如果获取到返回true,超时或者没有获取到返回false
         * 
         * @param timeOutGet    获取锁的超时时间(单位: 秒,不设置为10秒)
         * @param timeOutUse    使用锁的超时时间(单位: 秒,不设置为30秒)
         * @return
         */
        @Override
        public boolean lock(long timeOutGet, long timeOutUse) {
            return lock(timeOutGet, timeOutUse, LockUtil.TIME_UNIT);
        }

        /**
         * 在给定的时间内获取锁,如果获取到返回true,超时或者没有获取到返回false
         * 
         * @param timeOutGet    获取锁的超时时间(单位: 秒,不设置为10秒)
         * @param timeOutUse    使用锁的超时时间(单位: 秒,不设置为30秒)
         * @param unit          锁的超时时间单位
         * @return
         */
        @Override
        public boolean lock(long timeOutGet, long timeOutUse, TimeUnit unit) {
            try {
                if (timeOutUse == LockUtil.TIMEOUT_LONG ? getLock() : getLock(timeOutUse)) {
                    return true;
                } else if (waitForLock(waitNode, timeOutGet, unit)) {
                    return true;
                }
                throw new TransException(TransErrorCode.THREAD_BUSY);
            } catch (Exception e) {
                // 获取锁失败
                lockFail(e);
            }
            return false;
        }

        /**
         * 获取锁,如果获取到返回true,没有获取到返回false
         * 
         * @return
         * @throws Exception
         */
        private boolean getLock() throws Exception {
            // 处理锁
            getLockInit();

            // 创建临时子节点
            myNode = getZk().createTemporarySort(getLockNameAndSuffix(lockName));

            // 获取锁
            return getLockGo();
        }

        /**
         * 获取有超时时间的锁,如果获取到返回true,没有获取到返回false
         * 
         * @param timeOutUse    使用锁的超时时间(单位: 毫秒,不设置为30秒)
         * @return
         * @throws Exception
         */
        private boolean getLock(long timeOutUse) throws Exception {
            // 处理锁
            getLockInit();

            // 创建临时子节点
            myNode = getZk().createSort(getLockNameAndSuffix(lockName), timeOutUse * 1000);

            // 获取锁
            return getLockGo();
        }

        /**
         * 获取锁之前的处理
         * 
         * @return
         * @throws Exception
         */
        private void getLockInit() throws Exception {
            // if (lockName.contains(SPLITSTR)) {
            // throw new TransException("锁名称不能包含: " + SPLITSTR);
            // }
        }

        /**
         * 获取锁,如果获取到返回true,没有获取到返回false
         * 
         * @return
         * @throws Exception
         */
        private boolean getLockGo() throws Exception {
            ZookeeperUtilInterface zk = getZk();
            // 取出所有锁
            List<String> subNodes = zk.getChildren(root);

            // 取出所有lockName的锁
            List<String> lockObjNodes = new ArrayList<>();
            for (String node : subNodes) {
                // 用之前添加的分隔符分割出准确的锁名称
                String lockNameAllLeft = node.split(SPLITSTR)[0];
                // 判断是否是lockName锁
                if (lockNameAllLeft.equals(lockName)) {
                    lockObjNodes.add(node);
                }
            }
            // 将锁升序排序
            Collections.sort(lockObjNodes);

            // 判断当前锁是否是最小的锁
            if (myNode.equals(lockNameAllPath(lockObjNodes.get(0)))) {
                // 如果是最小的节点,则表示取得锁
                return true;
            }

            // 如果不是最小的锁,找到比自己小1的锁(获取锁加上后缀和索引后的名称)
            String submyNode = getLockNameAll(myNode);
            // 找到前一个锁
            waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, submyNode) - 1);
            // 返回false(没有拿到锁)
            return false;
        }

        /**
         * 获取锁失败后的处理
         * 
         * @param e
         */
        private void lockFail(Exception e) {
            // 删除临时节点并抛异常
            delete();
            throw new TransException(e.getMessage());
        }

        /**
         * 等待上一个锁执行完毕
         * 
         * @param lockNameAllPath   要等待的锁全路径
         * @param timeOutGet        获取锁的超时时间(单位: 秒,不设置为10秒)
         * @param unit           获取锁的超时时间单位
         * @return
         * @throws Exception
         */
        private boolean waitForLock(String lockNameAllPath, long timeOutGet, TimeUnit unit) throws Exception {
            // 设置一个计数器,如果上一个锁执行完毕了会让这个计数器减一,如果计数器为0计数器的await()方法就会恢复执行
            this.latch = new CountDownLatch(1);
            // 判断上一个锁是否存在,同时注册监听
            // (如果不存在就等于拿到锁了,如果存在就添加一个监听事件,当Zookeeper节点创建,删除,设置内容的时候会触发process()方法)
            Stat stat = getZk().exists(lockNameAllPath(lockNameAllPath), this);
            // 判断比自己小一个数的节点是否存在,如果存在就暂停线程
            if (stat != null) {
                // 让线程暂停一段时间
                if (timeOutGet == TIMEOUT_LONG) {
                    this.latch.await();
                } else {
                    this.latch.await(timeOutGet, unit);
                }
                // 不等于0表示超时了,但是还没有获取到锁,如果没有获取到就返回false
                if (this.latch.getCount() != 0) {
                    this.latch = null;
                    return false;
                }
            }

            this.latch = null;
            // 拿到锁了,返回true
            return true;
        }

        /**
         * 释放锁
         */
        @Override
        public void unlock() {
            // 删除临时节点
            delete();
        }

        @Override
        public void lockInterruptibly() throws InterruptedException {
            this.lock();
        }

        @Override
        public Condition newCondition() {
            return null;
        }

        /**
         * 获取锁加上后缀和索引后的名称
         * 
         * @param lockNameAllPath 锁全路径
         * @return
         */
        private String getLockNameAll(String lockNameAllPath) {
            return lockNameAllPath.substring(getRootLength());
        }

        /**
         * 获取锁全路径
         * 
         * @param lockNameAll 锁全称
         * @return
         */
        private String lockNameAllPath(String lockNameAll) {
            return root_ + lockNameAll;
        }

        /**
         * 获取锁没生成索引前的全路径
         * 
         * @param lockName 锁名称
         * @return
         */
        private String getLockNameAndSuffix(String lockName) {
            return new StringBuilder(root_).append(lockName).append(SPLITSTR).toString();
        }

        /**
         * 重新设置Zookeeper(防止临时节点删除失败,重启一个后之前会话的临时节点会消失)
         */
        @SuppressWarnings("unused")
        private void resetZookeeper() {
            try {
                getZk().close();
                setZk(createZk());
            } catch (Exception e) {
                log.error(e.getMessage());
                setZk(null);
                setZk(createZk());
            }
        }

        /**
         * 删除临时节点
         */
        private void delete() {
            try {
                getZk().delete(myNode);
                myNode = null;
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new TransException(e.getMessage());
            }
        }
    }

    /**
     * 获取Zookeeper工具类
     * 
     * @return
     */
    public ZookeeperUtilInterface createZk() {
        return ZookeeperUtil.getInstances().addAcl().open();
    }

    /**
     * 获取Zookeeper工具类
     * 
     * @return
     */
    public ZookeeperUtilInterface getZk() {
        return zk;
    }

    /**
     * 设置Zookeeper工具类
     * 
     * @param zk
     */
    public void setZk(ZookeeperUtilInterface zk) {
        this.zk = zk;
    }

    /**
     * 获取root路径的长度+1
     * 
     * @return
     */
    public int getRootLength() {
        return rootLength;
    }

    /**
     * 获取锁对象
     * 
     * @param lockName 锁名称
     * @return
     */
    @Override
    public CustomLock getLock(String lockName) {
        return new CustomLockWatcher(lockName);
    }

    /**
     * 判断一个对象是否是空(空对象)
     * 
     * @param obj
     * @return
     */
    private static boolean isBlankObject(Object obj) {
        return StringUtil.isBlankObject(obj);
    }

    public static DistributedLockZookeeperUtilImpl getInstances() {
        if (isBlankObject(dis)) {
            dis = new DistributedLockZookeeperUtilImpl();
        }
        return dis;
    }

}